package com.ilikesoup.instant.split;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.RecursiveAction;
import java.util.concurrent.RecursiveTask;
import java.util.function.*;

import static com.ilikesoup.instant.split.ForkJoinPoolUtils.parallelExecuteBatch;

/**
 * 分批执行工具，分页获取并执行工具
 * <p>
 * 在处理客户、员工等数据时，经常会调用ICE接口，通常在处理列表接口和分页接口时，查询数据量较大，会涉及到分批操作。
 * <p>
 * 该类旨在减少批量操作的代码，并提供并行分批处理的工具方法。
 * 在一些并行地增强方法（parallel*Pro）中，对共享变量的操作必须是线程安全的。
 * <p>
 * <p>
 * 1.用于基于列表批量操作，按批次进行划分和执行
 * 2.提供 串行 和 并行 (parallel*) 方法
 * 3.并行方法使用了ForkJoinPool
 */
public class BatchSplitUtils {

    private static final Logger LOGGER = LoggerFactory.getLogger(BatchSplitUtils.class);

    /**
     * 分批消费 List，无返回值
     *
     * @param list         源列表，不为null
     * @param listConsumer 列表消费函数，处理每一个批次
     * @param batchSize    每批大小
     * @param <T>          源列表元素类型
     */
    public static <T> void splitBatchAction(List<T> list, int batchSize, Consumer<List<T>> listConsumer) {
        if (list.isEmpty()) {
            return;
        }
        int times = (list.size() + batchSize - 1) / batchSize;
        for (int i = 0; i < times; i++) {
            if (i == times - 1) {
                listConsumer.accept(list.subList(i * batchSize, list.size()));
            } else {
                listConsumer.accept(list.subList(i * batchSize, (i + 1) * batchSize));
            }
        }
    }

    /**
     * 分页获取并分批消费 List，无返回值
     *
     * @param listProducer 根据页码和页长，获取处理列表
     * @param pageSize     页长，每页大小
     * @param listConsumer 列表消费函数，处理每一个批次
     * @param batchSize    每批大小
     * @param <T>          源列表元素类型
     */
    public static <T> void splitBatchPageAction(BiFunction<Integer, Integer, List<T>> listProducer, int pageSize, Consumer<List<T>> listConsumer, int batchSize) {
        doPageAction(listProducer, pageSize, pageList -> splitBatchAction(pageList, batchSize, listConsumer));
    }

    /**
     * 分页获取并消费 List，无返回值
     *
     * @param listProducer 根据页码和页长，获取处理列表
     * @param pageSize     页长，每页大小
     * @param listConsumer 列表消费函数，处理每一页
     * @param <T>          源列表元素类型
     */
    public static <T> void doPageAction(BiFunction<Integer, Integer, List<T>> listProducer, int pageSize, Consumer<List<T>> listConsumer) {
        int pageNo = 1;
        List<T> pageList;
        while ((pageList = listProducer.apply(pageNo++, pageSize)) != null && pageList.size() > 0) {
            listConsumer.accept(pageList);
        }
    }

    /**
     * 分批消费 List，无返回值
     * *增强：额外支持辅助参数
     *
     * @param list            源列表，不为null
     * @param listConsumerPro 列表消费函数，处理每一个批次
     * @param batchSize       批次大小
     * @param <T>             源列表元素类型
     * @param <K>             辅助参数类型
     */
    public static <T, K> void splitBatchActionPro(List<T> list, int batchSize, BiConsumer<List<T>, K> listConsumerPro, K param) {
        splitBatchAction(list, batchSize, subList -> listConsumerPro.accept(subList, param));
    }


    /**
     * 分批处理 List，有返回值，返回值要可以支持合并
     *
     * @param list      源列表，不为null
     * @param applyFunc 列表处理函数，处理每一个批次
     * @param joinFunc  合并函数，合并两个批次的返回值
     * @param batchSize 批次大小
     * @param <T>       源列表元素类型
     * @param <V>       返回值类型
     * @return 返回列表处理后的值
     */
    public static <T, V> V splitBatch(List<T> list, int batchSize, Function<List<T>, V> applyFunc, BinaryOperator<V> joinFunc) {
        if (list.size() <= batchSize) {
            return applyFunc.apply(list);
        }
        int times = (list.size() + batchSize - 1) / batchSize;
        V sum = applyFunc.apply(list.subList(0, batchSize));
        V temp;
        for (int i = 1; i < times; i++) {
            if (i == times - 1) {
                temp = applyFunc.apply(list.subList(i * batchSize, list.size()));
            } else {
                temp = applyFunc.apply(list.subList(i * batchSize, (i + 1) * batchSize));
            }
            sum = joinFunc.apply(sum, temp);
        }
        return sum;
    }

    /**
     * 分页获取并处理
     *
     * @param listProducer 根据页码和页长，获取处理列表
     * @param pageSize     页长，每页大小
     * @param applyFunc    列表处理函数，处理每一页，该函数可以使用该类中的其他方法，来分批处理结果。例如： pageList -> splitBatch(pageList, batchSize, applyFunction, joinFuncion)
     * @param joinFunc     合并函数，合并两个批次的返回值
     * @param <T>          源列表元素类型
     * @param <V>          返回值类型
     * @return 返回列表处理后的值
     */
    public static <T, V> V doPageTask(BiFunction<Integer, Integer, List<T>> listProducer, int pageSize, Function<List<T>, V> applyFunc, BinaryOperator<V> joinFunc) {
        int pageNo = 1;
        List<T> pageList;
        V result = null;
        while ((pageList = listProducer.apply(pageNo++, pageSize)) != null && pageList.size() > 0) {
            if (result == null) {
                result = applyFunc.apply(pageList);
            } else {
                result = joinFunc.apply(applyFunc.apply(pageList), result);
            }
        }
        return result;
    }

    /**
     * 分批处理 List，有返回值，返回值要可以支持合并
     * *增强：额外支持辅助参数
     *
     * @param list         源列表，不为null
     * @param applyProFunc 列表处理函数，处理每一个批次
     * @param joinFunc     合并函数，合并两个批次的返回值
     * @param batchSize    批次大小
     * @param <T>          源列表元素类型
     * @param <K>          辅助参数类型
     * @param <V>          返回值类型
     * @return 返回列表处理后的值
     */
    public static <T, K, V> V splitBatchPro(List<T> list, int batchSize, BiFunction<List<T>, K, V> applyProFunc, BinaryOperator<V> joinFunc, K param) {
        return splitBatch(list, batchSize, subList -> applyProFunc.apply(subList, param), joinFunc);
    }

    /**
     * 并行分批消费 List，无返回值
     *
     * @param taskName     任务名称，会在创建线程的时候写入线程名称
     * @param list         源列表，不为null
     * @param listConsumer 列表消费函数，处理每一个批次
     * @param batchSize    每批大小
     * @param <T>          源列表元素类型
     */
    public static <T> void parallelSplitBatchAction(String taskName, List<T> list, Consumer<List<T>> listConsumer, int batchSize) {
        if(list.isEmpty()) {
            return;
        }
        parallelExecuteBatch(new ListSplitAction<T>(list, listConsumer, batchSize), taskName);
    }

    /**
     * 并行单个消费 List，无返回值
     *
     * @param taskName     任务名称，会在创建线程的时候写入线程名称
     * @param list         源列表，不为null
     * @param consumer     单条消费函数，处理每一个
     * @param <T>          源列表元素类型
     */
    public static <T> void parallelSplitBatchAction(String taskName, List<T> list, Consumer<T> consumer) {
        if(list.isEmpty()) {
            return;
        }
        parallelExecuteBatch(new ListSplitAction<T>(list, subList -> consumer.accept(subList.get(0)), 1), taskName);
    }

    /**
     * 并行分批消费 List，无返回值
     * *增强：额外支持辅助参数
     *
     * @param taskName        任务名称，会在创建线程的时候写入线程名称
     * @param list            源列表，不为null
     * @param listConsumerPro 列表消费函数，处理每一个批次
     * @param batchSize       批次大小
     * @param <T>             源列表元素类型
     * @param <K>             辅助参数类型
     */
    public static <T, K> void parallelSplitBatchActionPro(String taskName, List<T> list, BiConsumer<List<T>, K> listConsumerPro, K param, int batchSize) {
        if(list.isEmpty()) {
            return;
        }
        parallelExecuteBatch(new ListSplitProAction<T, K>(list, batchSize, listConsumerPro, param), taskName);
    }

    /**
     * 并行分批处理 List，有返回值，返回值要可以支持合并
     *
     * @param taskName  任务名称，会在创建线程的时候写入线程名称
     * @param list      源列表，不为null
     * @param applyFunc 列表处理函数，处理每一个批次
     * @param joinFunc  合并函数，合并两个批次的返回值
     * @param batchSize 批次大小
     * @param <T>       源列表元素类型
     * @param <V>       返回值类型
     * @return 返回列表处理后的值
     */
    public static <T, V> V parallelSplitBatch(String taskName, List<T> list, Function<List<T>, V> applyFunc, int batchSize, BinaryOperator<V> joinFunc) {
        return parallelExecuteBatch(new ListSplitTask<T, V>(list, applyFunc, batchSize, joinFunc), taskName);
    }

    /**
     * 并行单个处理 List，有返回值，返回值要可以支持合并
     *
     * @param taskName  任务名称，会在创建线程的时候写入线程名称
     * @param list      源列表，不为null
     * @param applyFunc 单个处理函数，处理每一个
     * @param joinFunc  合并函数，合并两个批次的返回值
     * @param <T>       源列表元素类型
     * @param <V>       返回值类型
     * @return 返回列表处理后的值
     */
    public static <T, V> V parallelSplitBatch(String taskName, List<T> list, Function<T, V> applyFunc, BinaryOperator<V> joinFunc) {
        return parallelExecuteBatch(new ListSplitTask<T, V>(list, subList -> applyFunc.apply(subList.get(0)), 1, joinFunc), taskName);
    }

    /**
     * 并行分批处理 List，有返回值，返回值要可以支持合并
     * *增强：额外支持辅助参数
     *
     * @param taskName     任务名称，会在创建线程的时候写入线程名称
     * @param list         源列表，不为null
     * @param applyProFunc 列表处理函数，处理每一个批次
     * @param joinFunc     合并函数，合并两个批次的返回值
     * @param batchSize    批次大小
     * @param <T>          源列表元素类型
     * @param <K>          辅助参数类型
     * @param <V>          返回值类型
     * @return 返回列表处理后的值
     */
    public static <T, K, V> V parallelSplitBatchPro(String taskName, List<T> list, int batchSize, BinaryOperator<V> joinFunc, BiFunction<List<T>, K, V> applyProFunc, K sharePart) {
        return parallelExecuteBatch(new ListSplitProTask<T, K, V>(list, batchSize, joinFunc, applyProFunc, sharePart), taskName);
    }

    public static <T> void parallelSplitBatchPageAction(String taskName, BiFunction<Integer, Integer, List<T>> listProducer, int pageSize, Consumer<List<T>> listConsumer) {
        parallelExecuteBatch(new PageBatchAction<>(listProducer, pageSize, list -> new ListSplitAction<>(list, listConsumer, pageSize)), taskName);
    }

    public static <T> void parallelSplitBatchPageAction(String taskName, BiFunction<Integer, Integer, List<T>> listProducer, int pageSize, Consumer<List<T>> listConsumer, int batchSize) {
        parallelExecuteBatch(new PageBatchAction<>(listProducer, pageSize, list -> new ListSplitAction<>(list, listConsumer, batchSize)), taskName);
    }

    /**
     * 并行分页分批处理 List，有返回值，返回值要可以支持合并
     *
     * @param taskName     任务名称，会在创建线程的时候写入线程名称
     * @param listProducer pageNo, pageSize => list 通过分页参数查询列表函数
     * @param applyFunc    列表处理函数，处理每一个批次
     * @param joinFunc     合并函数，合并两个批次的返回值
     * @param batchSize    批次大小
     * @param <T>          源列表元素类型
     * @param <V>          返回值类型
     * @return 返回列表处理后的值
     */
    public static <T, V> V parallelSplitBatchPage(String taskName, BiFunction<Integer, Integer, PageList<T>> listProducer, int pageSize, Function<List<T>, V> applyFunc, int batchSize, BinaryOperator<V> joinFunc) {
        return parallelExecuteBatch(new PageBatchTask<>(listProducer, pageSize, list -> new ListSplitTask<>(list, applyFunc, batchSize, joinFunc)), taskName);
    }

    public static <T, K, V> V parallelSplitBatchPagePro(String taskName, BiFunction<Integer, Integer, PageList<T>> listProducer, int pageSize, Function<List<T>, V> applyFunc, int batchSize, BinaryOperator<V> joinFunc, BiFunction<List<T>, K, V> applyProFunc, K sharePart) {
        return parallelExecuteBatch(new PageBatchTask<>(listProducer, pageSize, list -> new ListSplitProTask<>(list, batchSize, joinFunc, applyProFunc, sharePart)), taskName);
    }

    public static abstract class CompoundSplitTask<T, V> extends BatchSplitTask<T, V> {

        private BatchSplitTaskProcessor<T, V> subTaskProcesser;

        public CompoundSplitTask(List<T> list, int batchSize, BinaryOperator<V> joinFunc, BatchSplitTaskProcessor<T, V> subTaskProcesser) {
            super(list, batchSize, joinFunc);
            this.subTaskProcesser = subTaskProcesser;
        }

        @Override
        protected V doCompute() {
            return subTaskProcesser.process(list).invoke();
        }
    }


    public static abstract class BatchSplitAction<T> extends RecursiveAction {
        /**
         * 待处理列表
         */
        protected final List<T> list;
        /**
         * 批次大小，每批可能小于或等于该数值
         */
        protected final int batchSize;

        public BatchSplitAction(List<T> list, int batchSize) {
            this.list = list;
            this.batchSize = batchSize <= 0 ? 1 : batchSize;
        }

        @Override
        protected void compute() {
            if (list.size() <= batchSize) {
                doCompute();
            } else {
                int times = (list.size() + batchSize - 1) / batchSize;
                List<RecursiveAction> taskList = new ArrayList<>(times);
                for (int i = 0; i < times; i++) {
                    if (i == times - 1) {
                        taskList.add(createSubTask(list.subList(i * batchSize, list.size()), batchSize));
                    } else {
                        taskList.add(createSubTask(list.subList(i * batchSize, (i + 1) * batchSize), batchSize));
                    }
                }
                invokeAll(taskList);
            }
        }

        /**
         * 实际执行计算的方法
         *
         * @return
         */
        protected abstract void doCompute();

        /**
         * 创建子任务的方法，根据具体实现创建子任务
         *
         * @param subList
         * @param batchSize
         * @return
         */
        protected abstract RecursiveAction createSubTask(List<T> subList, int batchSize);
    }

    public static class ListSplitAction<T> extends BatchSplitAction<T> {
        private final Consumer<List<T>> consumer;

        public ListSplitAction(List<T> list, Consumer<List<T>> consumer, int batchSize) {
            super(list, batchSize);
            this.consumer = consumer;
        }

        protected void doCompute() {
            consumer.accept(list);
        }

        @Override
        protected ListSplitAction createSubTask(List<T> subList, int batchSize) {
            return new ListSplitAction(subList, consumer, batchSize);
        }
    }

    public static class ListSplitProAction<T, K> extends BatchSplitAction<T> {
        private final BiConsumer<List<T>, K> biConsumer;
        /**
         * 共享部分，用于每次处理过程中，需要额外操作的变量
         * 在applyProFunc中对该变量的操作必须是线程安全的
         */
        private final K sharePart;

        public ListSplitProAction(List<T> list, int batchSize, BiConsumer<List<T>, K> biConsumer, K sharePart) {
            super(list, batchSize);
            this.biConsumer = biConsumer;
            this.sharePart = sharePart;
        }

        @Override
        protected void doCompute() {
            biConsumer.accept(list, sharePart);
        }

        @Override
        protected BatchSplitAction<T> createSubTask(List<T> list, int batchSize) {
            return new ListSplitProAction<>(list, batchSize, biConsumer, sharePart);
        }
    }

    /**
     * 用于将列表批量操作，分割成各个小部分
     *
     * @param <T> 列表元素的类型
     * @param <V> 处理结果的类型
     */
    public static abstract class BatchSplitTask<T, V> extends RecursiveTask<V> {
        /**
         * 待处理列表
         */
        protected final List<T> list;
        /**
         * 批次大小，每批可能小于或等于该数值
         */
        protected final int batchSize;
        /**
         * 子任务处理结果合并函数
         */
        protected final BinaryOperator<V> joinFunc;

        public BatchSplitTask(List<T> list, int batchSize, BinaryOperator<V> joinFunc) {
            this.list = list;
            this.batchSize = batchSize <= 0 ? 1 : batchSize;
            this.joinFunc = joinFunc;
        }

        @Override
        protected V compute() {
            if (list.size() <= batchSize) {
                return doCompute();
            } else {
                RecursiveTask<V> subTask1 = createSubTask(list.subList(0, list.size() / 2), batchSize, joinFunc);
                subTask1.fork();
                RecursiveTask<V> subTask2 = createSubTask(list.subList(list.size() / 2, list.size()), batchSize, joinFunc);
                subTask2.fork();
                return joinFunc.apply(subTask1.join(), subTask2.join());
            }
        }

        /**
         * 实际执行计算的方法
         *
         * @return
         */
        protected abstract V doCompute();

        /**
         * 创建子任务的方法，根据具体实现创建子任务
         *
         * @param subList
         * @param batchSize
         * @param joinFunc
         * @return
         */
        protected abstract RecursiveTask<V> createSubTask(List<T> subList, int batchSize, BinaryOperator<V> joinFunc);

        public int getBatchSize() {
            return batchSize;
        }

        public BinaryOperator<V> getJoinFunc() {
            return joinFunc;
        }
    }

    public static class ListSplitTask<T, V> extends BatchSplitTask<T, V> {
        private final Function<List<T>, V> applyFunc;

        public ListSplitTask(List<T> list, Function<List<T>, V> applyFunc, int batchSize, BinaryOperator<V> joinFunc) {
            super(list, batchSize, joinFunc);
            this.applyFunc = applyFunc;
        }

        protected V doCompute() {
            return applyFunc.apply(list);
        }

        @Override
        protected BatchSplitTask<T, V> createSubTask(List<T> subList, int batchSize, BinaryOperator<V> joinFunc) {
            return new ListSplitTask<>(subList, applyFunc, batchSize, joinFunc);
        }

    }

    public static class ListSplitProTask<T, K, V> extends BatchSplitTask<T, V> {
        private final BiFunction<List<T>, K, V> applyProFunc;
        /**
         * 共享部分，用于每次处理过程中，需要额外操作的变量
         * 在applyProFunc中对该变量的操作必须是线程安全的
         */
        private final K sharePart;

        public ListSplitProTask(List<T> list, int batchSize, BinaryOperator<V> joinFunc, BiFunction<List<T>, K, V> applyProFunc, K sharePart) {
            super(list, batchSize, joinFunc);
            this.applyProFunc = applyProFunc;
            this.sharePart = sharePart;
        }

        @Override
        protected V doCompute() {
            return applyProFunc.apply(list, sharePart);
        }

        @Override
        protected BatchSplitTask<T, V> createSubTask(List<T> list, int batchSize, BinaryOperator<V> joinFunc) {
            return new ListSplitProTask<>(list, batchSize, joinFunc, applyProFunc, sharePart);
        }
    }

    public static class PageList<T> {
        private final List<T> list;
        private final int totalCount;
        private static final PageList<?> EMPTY = new PageList(Collections.emptyList(), 0);

        public PageList(List<T> list, int totalCount) {
            this.list = list;
            this.totalCount = totalCount;
        }

        public List<T> getList() {
            return list;
        }

        public int getTotalCount() {
            return totalCount;
        }

        public boolean isEmpty() {
            return list == null || list.isEmpty();
        }

        public static <T> PageList<T> empty() {
            return (PageList<T>) EMPTY;
        }

    }

    public static class PageBatchTask<T, V> extends RecursiveTask<V> {
        // pageNo, pageSize => pageList
        private final BiFunction<Integer, Integer, PageList<T>> pageListProducer;
        private final BatchSplitTaskProcessor<T, V> batchSplitTaskProcessor;
        private final int pageSize;
        private final int startPage;
        private Supplier<V> defValGetter;

        public PageBatchTask(BiFunction<Integer, Integer, PageList<T>> pageListProducer, int pageSize, BatchSplitTaskProcessor<T, V> batchSplitTaskProcessor) {
            this(pageListProducer, 1, pageSize, batchSplitTaskProcessor);
        }

        public PageBatchTask(BiFunction<Integer, Integer, PageList<T>> pageListProducer, int startPage, int pageSize, BatchSplitTaskProcessor<T, V> batchSplitTaskProcessor) {
            this.pageListProducer = pageListProducer;
            this.batchSplitTaskProcessor = batchSplitTaskProcessor;
            this.pageSize = pageSize;
            this.startPage = startPage;
        }

        public PageBatchTask(BiFunction<Integer, Integer, PageList<T>> pageListProducer, int pageSize, BatchSplitTaskProcessor<T, V> batchSplitTaskProcessor, Supplier<V> defValGetter) {
            this(pageListProducer, 1, pageSize, batchSplitTaskProcessor, defValGetter);
        }

        public PageBatchTask(BiFunction<Integer, Integer, PageList<T>> pageListProducer, int startPage, int pageSize, BatchSplitTaskProcessor<T, V> batchSplitTaskProcessor, Supplier<V> defValGetter) {
            this.pageListProducer = pageListProducer;
            this.batchSplitTaskProcessor = batchSplitTaskProcessor;
            this.pageSize = pageSize;
            this.startPage = startPage;
            this.defValGetter = defValGetter;
        }

        @Override
        protected V compute() {
            int pageNo = startPage;
            PageList<T> pageList = pageListProducer.apply(pageNo++, pageSize);
            if (pageList == null || pageList.isEmpty()) {
                return getDefVal();
            }
            int totalPage = (pageList.getTotalCount() + pageSize - 1) / pageSize;
            BatchSplitTask<T, V> batchSplitTask = batchSplitTaskProcessor.process(pageList.getList());
            List<RecursiveTask<V>> taskList = new ArrayList<>();
            taskList.add(batchSplitTask);
            // 添加后面页的任务
            while (pageNo <= totalPage) {
                taskList.add(new PageBatchSubTask<>(pageListProducer, pageNo++, pageSize, batchSplitTaskProcessor));
            }
            Collection<RecursiveTask<V>> taskResults = invokeAll(taskList);

            BinaryOperator<V> joinFunc = batchSplitTask.getJoinFunc();
            V result = getDefVal();
            V taskVal = null;
            for (RecursiveTask<V> task : taskResults) {
                try {
                    taskVal = task.get();
                } catch (InterruptedException e) {
                    LOGGER.error("分页并行操作任务执行被打断", e);
                } catch (ExecutionException e) {
                    LOGGER.error("分页并行操作任务执行异常", e);
                }
                if (result == null) {
                    result = nullDef(taskVal);
                } else {
                    result = joinFunc.apply(result, nullDef(taskVal));
                }
            }
            return result;
        }

        private V getDefVal() {
            return defValGetter == null ? null : defValGetter.get();
        }

        private V nullDef(V target) {
            return target == null ? getDefVal() : target;
        }

    }

    public static class PageBatchSubTask<T, V> extends RecursiveTask<V> {
        private final BiFunction<Integer, Integer, PageList<T>> pageListProducer;
        private final BatchSplitTaskProcessor<T, V> batchSplitTaskProcessor;
        private final int pageSize;
        private final int pageNo;

        public PageBatchSubTask(BiFunction<Integer, Integer, PageList<T>> pageListProducer, int pageNo, int pageSize, BatchSplitTaskProcessor<T, V> batchSplitTaskProcessor) {
            this.pageListProducer = pageListProducer;
            this.batchSplitTaskProcessor = batchSplitTaskProcessor;
            this.pageSize = pageSize;
            this.pageNo = pageNo;
        }

        @Override
        protected V compute() {
            PageList<T> pageList = pageListProducer.apply(pageNo, pageSize);
            if (pageList == null || pageList.isEmpty()) {
                return null;
            }
            return batchSplitTaskProcessor.process(pageList.getList()).invoke();
        }
    }


    public static class PageBatchAction<T> extends RecursiveAction {
        // pageNo, pageSize => pageList
        private final BiFunction<Integer, Integer, List<T>> listProducer;
        private final BatchSplitActionProcessor<T> batchSplitActionProcessor;
        private final int pageSize;
        private final int startPage;

        public PageBatchAction(BiFunction<Integer, Integer, List<T>> listProducer, int pageSize, BatchSplitActionProcessor<T> batchSplitActionProcessor) {
            this(listProducer, 1, pageSize, batchSplitActionProcessor);
        }

        public PageBatchAction(BiFunction<Integer, Integer, List<T>> listProducer, int startPage, int pageSize, BatchSplitActionProcessor<T> batchSplitActionProcessor) {
            this.listProducer = listProducer;
            this.batchSplitActionProcessor = batchSplitActionProcessor;
            this.pageSize = pageSize;
            this.startPage = startPage;
        }

        @Override
        protected void compute() {
            int pageNo = startPage;
            List<T> pageList;
            List<RecursiveAction> taskList = new ArrayList<>();
            while ((pageList = listProducer.apply(pageNo++, pageSize)) != null && pageList.size() > 0) {
                taskList.add(batchSplitActionProcessor.process(pageList));
            }
            if (taskList.isEmpty()) {
                return;
            }
            invokeAll(taskList);
        }

    }

    @FunctionalInterface
    public interface BatchSplitTaskProcessor<T, V> {
        BatchSplitTask<T, V> process(List<T> list);
    }

    @FunctionalInterface
    public interface BatchSplitActionProcessor<T> {
        BatchSplitAction<T> process(List<T> list);
    }

}
